package io.kiki.sba.registry.server.metadata.monitor.impl;


import io.kiki.sba.registry.common.model.slot.BaseSlotStatus;
import io.kiki.sba.registry.common.model.slot.FollowerSlotStatus;
import io.kiki.sba.registry.common.model.slot.LeaderSlotStatus;
import io.kiki.sba.registry.common.model.slot.Slot;
import io.kiki.sba.registry.server.metadata.monitor.SlotStats;
import org.apache.commons.lang.StringUtils;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;


public class SlotStatsImpl implements SlotStats {

    private final Slot slot;
    private final long maxSyncGap;
    /**
     * offsets: follower replicate offset against the leader. Current implementation is as simple as
     * no incremental data replication Under this circumstances, offset stands for the last successful
     * sync task execute time
     */
    private final Map<String, Long> dataServerToFollowerLastSyncTimeMap = new ConcurrentHashMap<>();
    private volatile BaseSlotStatus.LeaderStatus leaderStatus = BaseSlotStatus.LeaderStatus.init;


    public SlotStatsImpl(Slot slot, long maxSyncGap) {
        this.slot = slot;
        this.maxSyncGap = maxSyncGap;
    }


    @Override
    public Slot getSlot() {
        return slot;
    }


    @Override
    public boolean isLeaderStable() {
        return leaderStatus.isHealthy();
    }

    @Override
    public boolean isFollowersStable() {
        for (String dataServer : slot.getFollowerDataStoreNodeIdSet()) {
            if (!isFollowerStable(dataServer)) {
                return false;
            }
        }
        return true;
    }

    /**
     * Is follower stable boolean.
     *
     * @param dataServer the data server
     *                   <p>Calculate the replicate gap between leader and followers, check if it's legal Current
     *                   implementation is as simple as no incremental data replication Under this circumstances,
     *                   offset stands for the last successful sync task execute time
     * @return the boolean
     */
    @Override
    public boolean isFollowerStable(String dataServer) {
        if (StringUtils.isEmpty(dataServer)) {
            return false;
        }
        Long offset = dataServerToFollowerLastSyncTimeMap.get(dataServer);
        return offset != null && System.currentTimeMillis() - offset < maxSyncGap;
    }


    @Override
    public void updateLeaderState(LeaderSlotStatus leaderSlotStatus) {
        this.leaderStatus = leaderSlotStatus.getLeaderStatus();
    }

    @Override
    public void updateFollowerState(FollowerSlotStatus followerSlotStatus) {
        dataServerToFollowerLastSyncTimeMap.put(followerSlotStatus.getServer(), followerSlotStatus.getLastLeaderSyncTime());
    }


    @Override
    public boolean equals(Object o) {
        if (this == o)
            return true;
        if (o == null || getClass() != o.getClass())
            return false;
        SlotStatsImpl that = (SlotStatsImpl) o;
        return Objects.equals(slot, that.slot) && leaderStatus == that.leaderStatus && Objects.equals(dataServerToFollowerLastSyncTimeMap, that.dataServerToFollowerLastSyncTimeMap);
    }

    @Override
    public int hashCode() {
        return Objects.hash(slot, leaderStatus, dataServerToFollowerLastSyncTimeMap);
    }
}
